package com.heima.article.listeren;

import com.alibaba.fastjson.JSON;
import com.heima.model.article.ArticleVisitStreamMess;
import com.heima.model.article.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;
import sun.security.provider.MD4;

import javax.lang.model.util.AbstractTypeVisitor6;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        // 创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        stream.map((key, value) -> {
            if (StringUtils.isEmpty(value)) {
                return new KeyValue<>(key, value);
            }
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);

            return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
        })
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
        .aggregate(
                new Initializer<String>() {

                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                },
                new Aggregator<String, String, String>() {
                    @Override
                    public String apply(String key, String value, String initData) {
                        String[] split = value.split(":");
                        String type = split[0];
                        String add = split[1];

                        Map<String, Integer> map =intData2Map(initData);
                        Integer newValue = map.get(type) + Integer.parseInt(add);
                        map.put(type, newValue);

                        return map2String(map);
                    }
                },
                Materialized.as("123")
        ).toStream()
         .map((key,value) ->{
             return new KeyValue<>(key.key().toString(),string2JOSN(value,Long.parseLong(key.key())));
        }).to("itcast-topic-output");
        return stream;
    }
    private Map<String,Integer> intData2Map(String initData){
        if (StringUtils.isBlank(initData)){
            return new HashMap<>();
        }
        HashMap<String, Integer> map = new HashMap<>();
        String[] split = initData.split(",");
        if (ArrayUtils.isEmpty(split)){
            return new HashMap<>();
        }
        for (String item : split) {
            if (StringUtils.isBlank(item)){
                continue;
            }
            String[] split1 = item.split(":");
            String key = split1[0];
            String value = split1[1];
            map.put(key,Integer.parseInt(value));
        }
        return map;
    }

    private String map2String(Map<String,Integer> map){
        Set<String> set = map.keySet();
        if (CollectionUtils.isEmpty(set)){
            return "";
        }
        StringBuilder builder=new StringBuilder();
        for (String key : set) {
            if (StringUtils.isBlank(key)){
                continue;
            }
            Integer value = map.get(key);
            builder.append(key).append(":").append(value).append(",");
        }
        String str = builder.toString();
        return str.substring(0,str.length()-1);
    }

    private String string2JOSN(String str, Long articleId){
        Map<String, Integer> map = intData2Map(str);
        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
        articleVisitStreamMess.setArticleId(articleId);
        articleVisitStreamMess.setCollect(map.get("COLLECTION"));
        articleVisitStreamMess.setComment(map.get("COMMENT"));
        articleVisitStreamMess.setLike(map.get("LIKES"));
        articleVisitStreamMess.setView(map.get("VIEWS"));
        return JSON.toJSONString(articleVisitStreamMess);
    }
}